//package com.example.service;
//
//import com.example.Utils.Env;
//import com.example.entity.StockTrade;
//import com.example.entity.TradeResult;
//import jakarta.annotation.PostConstruct;
//import jakarta.annotation.Resource;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.flink.api.common.functions.AggregateFunction;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
//import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
//import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
//import org.apache.flink.streaming.api.windowing.time.Time;
//import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
//import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
//import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
//import org.apache.flink.util.Collector;
//import org.springframework.stereotype.Component;
//import redis.clients.jedis.Jedis;
//
//import java.util.HashMap;
//import java.util.Map;
//
//@Component
//@Slf4j
//public class StockTradeAnalysis {
//    @Resource
//    private Env env_;
//
//    @PostConstruct
//    public void init() {
//        new Thread(() -> {
//            try {
//                startFlinkJob();
//            } catch (Exception e) {
//                log.error("Error starting Flink job", e);
//            }
//        }).start();
//    }
//
//    public  void startFlinkJob() throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        DataStream<StockTrade> tradeStream = env_.getEnv(env);
//        log.info("数据源搞好了");
//
//        // 1 小时窗口
//        tradeStream
//                .windowAll(TumblingEventTimeWindows.of(Time.hours(1)))
//                .trigger(EventTimeTrigger.create())
//                .aggregate(new GlobalTradeAggregator(), new GlobalTradeResultFunction("1h"))
//                .addSink(new AccumulatingRedisSink(redisConfig));
//
//        env.execute("Stock Trade Analysis");
//    }
//
//    public FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder()
//            .setHost("localhost")
//            .setPort(6379)
//            .build();
//
//    // 自定义Sink函数
//    public static class AccumulatingRedisSink extends RichSinkFunction<TradeResult> {
//        private transient Jedis jedis;
//        private final FlinkJedisPoolConfig config;
//
//        public AccumulatingRedisSink(FlinkJedisPoolConfig config) {
//            this.config = config;
//        }
//
//        @Override
//        public void open(Configuration parameters) throws Exception {
//            super.open(parameters);
//            jedis = new Jedis(config.getHost(), config.getPort());
//        }
//
//        @Override
//        public void close() throws Exception {
//            if (jedis != null) {
//                jedis.close();
//            }
//            super.close();
//        }
//
//        @Override
//        public void invoke(TradeResult value, Context context) throws Exception {
//            String existingValue = jedis.get("1:");
//
//            double totalAmount;
//            double totalVolumeCount;
//
//            if (existingValue != null) {
////                System.out.println("Existing Value: " + existingValue);
//                String[] parts = existingValue.split(":");
//                totalAmount = Double.parseDouble(parts[0]);
//                totalVolumeCount = Double.parseDouble(parts[1]);
//            } else {
//                System.out.println("我是null");
//                totalAmount = 0.0;
//                totalVolumeCount = 0L;
//            }
//
//            totalAmount += Double.parseDouble(value.getCount());
//            totalVolumeCount += Double.parseDouble(value.getTradeVolume());
////            System.out.println("Total Amount: " + totalAmount);
////            System.out.println("Total Volume Count: " + totalVolumeCount);
//
//            jedis.set("1:", totalAmount + ":" + totalVolumeCount);
//            final long delayInterval = 100;
//            Thread.sleep(delayInterval);
//        }
//    }
//
//    // 定义全局聚合函数
//    public static class GlobalTradeAggregator implements AggregateFunction<StockTrade, Map<String, TradeAccumulator>, Map<String, TradeAccumulator>> {
//        @Override
//        public Map<String, TradeAccumulator> createAccumulator() {
//            return new HashMap<>();
//        }
//
//        @Override
//        public Map<String, TradeAccumulator> add(StockTrade value, Map<String, TradeAccumulator> accumulator) {
//            String stockCode = value.getStockCode();
//            TradeAccumulator acc = accumulator.getOrDefault(stockCode, new TradeAccumulator(stockCode));
//            acc.totalAmount += value.getTradeVolume();
//            acc.totalVolumeCount++;
//
//            accumulator.put(stockCode, acc);
//
//            return accumulator;
//        }
//
//        @Override
//        public Map<String, TradeAccumulator> getResult(Map<String, TradeAccumulator> accumulator) {
//            TradeAccumulator.type++;
//
//            return accumulator;
//        }
//
//        @Override
//        public Map<String, TradeAccumulator> merge(Map<String, TradeAccumulator> a, Map<String, TradeAccumulator> b) {
//            for (Map.Entry<String, TradeAccumulator> entry : b.entrySet()) {
//                String stockCode = entry.getKey();
//                TradeAccumulator accB = entry.getValue();
//                TradeAccumulator accA = a.getOrDefault(stockCode, new TradeAccumulator(stockCode));
//                accA.totalAmount += accB.totalAmount;
//                accA.totalVolumeCount += accB.totalVolumeCount;
//                a.put(stockCode, accA);
//            }
//            return a;
//        }
//    }
//
//    // 定义全局结果函数
//    public static class GlobalTradeResultFunction implements AllWindowFunction<Map<String, TradeAccumulator>, TradeResult, TimeWindow> {
//        private final String windowType;
//
//        public GlobalTradeResultFunction(String windowType) {
//            this.windowType = windowType;
//        }
//
//        @Override
//        public void apply(TimeWindow window, Iterable<Map<String, TradeAccumulator>> input, Collector<TradeResult> out) throws Exception {
//            for (Map<String, TradeAccumulator> accumulatorMap : input) {
//                for (Map.Entry<String, TradeAccumulator> entry : accumulatorMap.entrySet()) {
//                    String stockCode = entry.getKey();
//                    TradeAccumulator accumulator = entry.getValue();
//                    log.info("Processing window for stock code: {}, window type: {}, window: [{} - {}], total amount: {}, total volume count: {}，这是第{}次",
//                            stockCode, windowType, window.getStart(), window.getEnd(), accumulator.totalAmount, accumulator.totalVolumeCount,TradeAccumulator.type);
//                    out.collect(new TradeResult(stockCode, String.valueOf(accumulator.totalAmount), String.valueOf(accumulator.totalVolumeCount)));
//                }
//            }
//        }
//    }
//
//    // 定义累加器
//    public static class TradeAccumulator {
//        public static int type = 0;
//        public String stockCode;
//        public double totalAmount = 0.0;
//        public long totalVolumeCount = 0L;
//
//        public TradeAccumulator(String stockCode) {
//            this.stockCode = stockCode;
//        }
//    }
//}
